
#include <apr_atomic.h>
#include <apr_strings.h>
#include "../aprtools/APRAutoLock.h"
#include "../aprtools/APRFile.h"
#include "../iface/APRInterface.h"
#include "../iface/debug.h"
#include "../utils/Hash.h"
#include "../utils/BitVector.h"
#include "BlockWriteServer.h"

/** Constructor for the block writing server.
 * 
 * @param inoofthreads The number of threads to use for serving
 * connections.
 * 
 * @param iportno The port to listen on.
 * 
 * @param ihostname The host (interface) to listen on.
 * 
 * @param upperpool The memory pool to derive this server memory pool
 * from.
 * 
 * @param irportno If we are behind a firewall or NAT or somesuch,
 * this is the remote port for the punching service.
 * 
 * @param irhostname If we are behind a firewall or NAT or somesuch,
 * this is the remote hostname for the punching service.
 * 
 * @param inomemfileblocks The maximum number of possible active
 * blocks or active slots to initialise at one time. */
BlockWriteServer::BlockWriteServer(int inoofthreads, int iportno, const char *ihostname, apr_pool_t *upperpool, int irportno, const char *irhostname, int inomemfileblocks)
 : UDPServer(inoofthreads, iportno, ihostname, upperpool)
{
   rportno = irportno;
   rhostname = (irhostname) ? strdup(irhostname) : NULL;
   nomemfileblocks = inomemfileblocks;
   blockarray = NULL;
   incidx = 0;
   blockinclock = NULL;
   watchthreadstop = 0;
}

/** Destructor for the block writing server. */
BlockWriteServer::~BlockWriteServer()
{
   joinWatchThread();
   if (rhostname) free(rhostname);
   if (blockinclock) delete blockinclock;
   blockinclock = NULL;
   if (watchthread) delete watchthread;
   watchthread = NULL;
   if (fileslock) delete fileslock;
   fileslock = NULL;
}

/** Start handling clients to the server (the client handler threads
 * run this).
 * 
 * @param threadid The integer thread ID number.
 * 
 * @param localmp The memory pool local to this thread. */
void BlockWriteServer::handleClientsOnThread(int threadid, apr_pool_t *localmp)
{
   while (!sessionstop)
   {
      pipearray[threadid].sessionstart->wait();
      if (sessionstop) break;
      apr_pool_t *sp;
      apr_pool_create(&sp, mp);
      int has_read = 0, has_written = 0, has_finalised = 0;
      if (pipearray[threadid].datagram_length == (PACKET_BLOCK_SIZE + (sizeof(uint32_t) * 2)))
      {
         // blockID, offset, data
         uint32_t p_blockID, p_offset;
         p_blockID = ((uint16_t *)(pipearray[threadid].datagram_data))[0];
         p_offset = ((uint16_t *)(pipearray[threadid].datagram_data))[1];
         const char *p_data = (const char *)(&(((uint32_t *)(pipearray[threadid].datagram_data))[1]));
         writePacket(p_blockID, p_offset * PACKET_BLOCK_SIZE, p_data);
         printf("%s\n", pipearray[threadid].datagram_data);
         pipearray[threadid].complete = 1;
      }
      apr_pool_clear(sp);
      apr_pool_destroy(sp);
   }
   pipearray[threadid].sessionstart->unlock();
}

/** Start handling clients to the server (the client handler threads
 * run this).
 * 
 * @param tthread The APR pointer to this thread grabbed.
 * 
 * @param args The void pointer given to the thread. This is actually
 * a pointer to an array of two void pointers, which are the this
 * pointer and a pointer to the thread ID respectively.
 * 
 * @return Return is NULL. */
void *BlockWriteServer::handleClientsOnThreadStatic(apr_thread_t *tthread, void *args)
{
   apr_status_t status;
   BlockWriteServer *kthis = *((BlockWriteServer **)(((void **)(args))[0]));
   int i = *((int *)(((void **)(args))[1]));
   apr_pool_t *upperpool, *mp;
   upperpool = apr_thread_pool_get(tthread);
   status = apr_pool_create(&mp, upperpool);
   CheckAPRError(status);

   kthis->handleClientsOnThread(i, mp);

   apr_pool_clear(mp);
   apr_pool_destroy(mp);
   ERR_remove_state(0);
   apr_thread_exit(tthread, APR_SUCCESS);
   return NULL;
}

/** Start watching over the active blocks (this is a separate thread).
 * 
 * @param localmp The local memory pool to this thread that is
 * destroyed afterwards. */
void BlockWriteServer::watchFromThread(apr_pool_t *localmp)
{
   while (!watchthreadstop)
   {
      // Need to keep an eye on things to ensure no UDP blocks last
      // too long and block things up.
      for (int i = 0; i < nomemfileblocks; i++)
      {
         if ((blockarray[i].active) && (blockarray[i].packetcounter == 0))
         {
            // Ensure that this slot is shut down
            stopFileBlock(i);
         }
         apr_atomic_set32(&(blockarray[i].packetcounter), 0);
      }
      apr_sleep(10000000);
   }
}

/** Start watching over the active blocks (this is a separate thread).
 * 
 * @param tthread The APR pointer to this thread grabbed.
 * 
 * @param args The void pointer given to the thread. This is actually
 * a pointer to the this pointer.
 * 
 * @return Return is NULL. */
void *BlockWriteServer::watchFromThreadStatic(apr_thread_t *tthread, void *args)
{
   apr_status_t status;
   BlockWriteServer *kthis = (BlockWriteServer *)args;
   apr_pool_t *upperpool, *mp;
   upperpool = apr_thread_pool_get(tthread);
   apr_pool_create(&mp, upperpool);
   
   kthis->watchFromThread(mp);

   ERR_remove_state(0);
   apr_pool_destroy(mp);
   status = apr_thread_exit(tthread, APR_SUCCESS);
   CheckAPRError(status);
   return NULL;
}

/** Start watching thread. The watching thread is intended to shut
 * down any unused, completed or idle slots so that others may use the
 * limited number of resources. */
int BlockWriteServer::startWatchThread()
{
   blockinclock = new APRMutex(mp);
   watchthread = new APRMutex(mp);
   fileslock = new APRMutex(mp);
   blockarray = (BlockActiveBlock *)apr_palloc(mp, sizeof(BlockActiveBlock)*nomemfileblocks);
   for (int i = 0; i < nomemfileblocks; i++)
   {
      apr_atomic_set32(&(blockarray[i].blockid), 0);
      apr_atomic_set32(&(blockarray[i].active), 0);
      apr_atomic_set32(&(blockarray[i].packetcounter), 0);
   }
   apr_status_t status;
   status = apr_threadattr_create(&watch_threadattr, mp);
   CheckAPRError(status);
   status = apr_thread_create(&watch_thread, watch_threadattr, (functor)(&(BlockWriteServer::watchFromThreadStatic)), (void *)this, mp);
   CheckAPRError(status);
   return 0;
}

/** Join watching thread. Stops the watching thread, and waits for it
 * to join the main thread. */
int BlockWriteServer::joinWatchThread()
{
   watchthreadstop = 1;
   apr_status_t status, thread_status;
   status = apr_thread_join(&thread_status, watch_thread);
   CheckAPRError(status);
   return 0;
}

/** Initialises an active slot to contain a file transfer block in
 * progress.  This gives a workable slot with an ID that can be
 * transferred to.
 * 
 * @param filename File name to open as the end point of the transfer
 * (to be overwritten with incoming data).
 * 
 * @param offset The start offset through the file of the active block
 * that will be represented by the active slot.
 * 
 * @param block_length The length of the block in the file to be
 * transferred to.
 * 
 * @param goodhashes An array of raw binary hashes for the data. These
 * hashes are to be compared with the hashed blocks to verify them.
 * 
 * @param blockid_result A pointer that will be written into with the
 * block ID when the file block is started successfully.
 * 
 * @return A return code that is zero on success and gives a non-zero
 * return code on failure. */
int BlockWriteServer::startFileBlock(const char *filename, uint64_t offset, int64_t block_length, char *goodhashes, uint32_t *blockid_result)
{
   apr_status_t status;
   APRAutoLock autofileslock(fileslock);
   if (!filename)
   {
      errprint("file name NULL");
      return BLOCK_WRITE_SERVER__FILE_NAME_ERROR;
   }
   // If the block length requested is too big, reject the request
   if (block_length > ACTIVE_BLOCK_SIZE)
   {
      errprint("requested block size %" APR_INT64_T_FMT " too big (bigger than %d)", block_length, ACTIVE_BLOCK_SIZE);
      return BLOCK_WRITE_SERVER__REQUESTED_BLOCK_TOO_BIG;
   }
   int slot_to_use = -1;
   uint32_t this_idx = 0;
   // Select the slot to use and mark it in use behind a lock
   if (1)
   {
      APRAutoLock autolock(blockinclock);
      // Select the block ID number - make sure it isn't zero.
      this_idx = ((incidx & 0xFFFF) != 0) ? (incidx) : (++incidx);
      incidx++;
      for (int i = 0; i < nomemfileblocks; i++)
      {
         // Set a free block as active and exit
         if (!(blockarray[i].active))
         {
            blockarray[i].active = 1;
            status = apr_pool_create(&(blockarray[i].localmp), mp);
            CheckAPRError(status);
            slot_to_use = i;
            break;
         }
      }
   }
   // If no slots have been found, then they must all be in use. Set
   // inactive. Return error.
   if (slot_to_use == -1)
   {
      errprint("can't start new memory file block, all block slots in use");
      return BLOCK_WRITE_SERVER__ALL_SLOTS_IN_USE;
   }
   // Open the target file
   apr_file_t *filepntr;
   status = APRFile::openWrapper(&filepntr, filename, APR_FOPEN_BINARY|APR_FOPEN_READ|APR_FOPEN_BUFFERED, APR_FPROT_OS_DEFAULT, blockarray[slot_to_use].localmp);
   // If we can't open it, it probably doesn't exist. Set
   // inactive. Return error.
   if (status != APR_SUCCESS)
   {
      errprint("file '%s' unallocated, not allowed", filename);
      apr_pool_destroy(blockarray[slot_to_use].localmp);
      blockarray[slot_to_use].active = 0;
      return BLOCK_WRITE_SERVER__FILE_UNALLOCATED;
   }
   // Find the end of the file.
   apr_off_t fileendtobe = 0;
   status = apr_file_seek(filepntr, APR_END, &fileendtobe);
   CheckAPRError(status);
   // If the block begins beyond the end of the file - this is
   // inoperable, set inactive, return error.
   if (offset >= fileendtobe)
   {
      errprint("file '%s': block begins (%" APR_SIZE_T_FMT ") beyond end of file (%" APR_OFF_T_FMT ")", filename, offset, fileendtobe);
      apr_pool_destroy(blockarray[slot_to_use].localmp);
      blockarray[slot_to_use].active = 0;
      return BLOCK_WRITE_SERVER__BLOCK_BEYOND_FILE_END;
   }
   // If the block ends beyond the end of the file, we trim the block
   // size.
   int64_t real_block_length = ((fileendtobe - offset) > block_length) ? block_length : (fileendtobe - offset);
   int32_t this_length = 0;
   int q = 0;
   // Go through each hashed block. Load from file.
   for (int i = 0; i < real_block_length; i+=HASHING_BLOCK_SIZE)
   {
      // Seek to the appropriate point in the file
      apr_off_t thisseekd = offset + i;
      status = apr_file_seek(filepntr, APR_SET, &thisseekd);
      CheckAPRError(status);
      if (status != APR_SUCCESS)
      {
         errprint("could not seek to %" APR_OFF_T_FMT " in file '%s'", (apr_off_t)(offset + i), filename);
         apr_pool_destroy(blockarray[slot_to_use].localmp);
         blockarray[slot_to_use].active = 0;
         return BLOCK_WRITE_SERVER__FILE_SEEK_ERROR;
      }
      // Read block from file
      this_length = ((real_block_length - i) > HASHING_BLOCK_SIZE) ? HASHING_BLOCK_SIZE : (real_block_length - i);
      apr_size_t fullread = this_length;
      status = apr_file_read_full(filepntr, blockarray[slot_to_use].hashblock[q].hashblockdata, this_length, &fullread);
      CheckAPRError(status);
      // If reading fails return error.
      if (status != APR_SUCCESS)
      {
         errprint("could not read block bytes %" APR_OFF_T_FMT "-%" APR_OFF_T_FMT, (apr_off_t)(offset + i), (apr_off_t)(offset + i + this_length));
         apr_pool_destroy(blockarray[slot_to_use].localmp);
         blockarray[slot_to_use].active = 0;
         return BLOCK_WRITE_SERVER__FILE_READ_ERROR;
      }
      // Pad a block that doesn't finish with zeroes.
      for (int i = this_length; i < HASHING_BLOCK_SIZE; i++)
         blockarray[slot_to_use].hashblock[q].hashblockdata[i] = 0x00;
      // Block out the filled vector parts that are outside the block
      // with 1.
      for (int j = ((this_length / PACKET_BLOCK_SIZE) + 1); j < (HASHING_BLOCK_SIZE / PACKET_BLOCK_SIZE); j++)
         BitVector::setBitFast(blockarray[slot_to_use].hashblock[q].filled_bit_vector, j);
      if ((this_length < HASHING_BLOCK_SIZE) && ((this_length % PACKET_BLOCK_SIZE) == 0))
         BitVector::setBitFast(blockarray[slot_to_use].hashblock[q].filled_bit_vector, (this_length / PACKET_BLOCK_SIZE));
      // Also add the appropriate number of packets to the packet
      // count
      blockarray[slot_to_use].hashblock[q].committedblocks = (HASHING_BLOCK_SIZE - this_length) / PACKET_BLOCK_SIZE;
      // Put a SHA-1 hash of the data into the checksum part of the
      // hash block.
      Hash::getBinarySHA1Hash((const char *)(blockarray[slot_to_use].hashblock[q].hashblockdata), HASHING_BLOCK_SIZE, blockarray[slot_to_use].hashblock[q].checksum);
      q++;
   }
   // Fill in verified tables for blocks beyond the end of the real
   // block length
   for (int i = ((real_block_length / HASHING_BLOCK_SIZE) + 1); i < (ACTIVE_BLOCK_SIZE / HASHING_BLOCK_SIZE); i++)
      BitVector::setBitFast(blockarray[slot_to_use].verified_bit_vector, i);
   if ((real_block_length < ACTIVE_BLOCK_SIZE) && ((real_block_length % HASHING_BLOCK_SIZE) == 0))
      BitVector::setBitFast(blockarray[slot_to_use].verified_bit_vector, (real_block_length / HASHING_BLOCK_SIZE));
   // Close the file
   status = apr_file_close(filepntr);
   CheckAPRError(status);
   blockarray[slot_to_use].filename = (char *)apr_pstrdup(blockarray[slot_to_use].localmp, filename);
   blockarray[slot_to_use].hashes = (uint8_t *)apr_palloc(blockarray[slot_to_use].localmp, sizeof(uint8_t)*SHA1_DIGEST_LENGTH*(ACTIVE_BLOCK_SIZE / HASHING_BLOCK_SIZE));
   // Memcopy hashes into array
   memcpy(blockarray[slot_to_use].hashes, goodhashes, sizeof(uint8_t)*SHA1_DIGEST_LENGTH*(ACTIVE_BLOCK_SIZE / HASHING_BLOCK_SIZE));
   blockarray[slot_to_use].offset = offset;
   blockarray[slot_to_use].blocklength = block_length;
   // Copy in and compare known good hashes:
   q = 0;
   // Go through each hashed block. Compare the hashes to the
   // goodhashes.
   for (int i = 0; i < real_block_length; i+=HASHING_BLOCK_SIZE)
   {
      // Test to see if the hash given matches the computed hash.
      int test = (memcmp(blockarray[slot_to_use].hashblock[q].checksum, &(goodhashes[q*SHA1_DIGEST_LENGTH]), sizeof(char)*SHA1_DIGEST_LENGTH) == 0);
      // Compare and set, if memcmp is the SAME, test evaluates to 1
      // and casBit sets the 'hash verified' bit.  Otherwise, the bit
      // is unset.
      BitVector::setCondBitFast(blockarray[slot_to_use].verified_bit_vector, test, q);
      // Fill in all bit vector if it matches the checksum, or else
      // blank all bit vector.
      for (int j = 0; j < ((HASHING_BLOCK_SIZE / PACKET_BLOCK_SIZE) / (sizeof(uint32_t) * 8)); j++)
         blockarray[slot_to_use].hashblock[q].filled_bit_vector[j] = ((test) ? (~(uint32_t)0) : 0);
      // Say all the blocks are already here if the hash test
      // succeeds.
      if (test)
         blockarray[slot_to_use].hashblock[q].committedblocks = (HASHING_BLOCK_SIZE / PACKET_BLOCK_SIZE);
      q++;
   }
   apr_atomic_set32(&(blockarray[slot_to_use].packetcounter), 0);
   apr_atomic_set32(&(blockarray[slot_to_use].blockid), this_idx);
   if (blockid_result) *blockid_result = this_idx;
   return BLOCK_WRITE_SERVER__SUCCESS;
}

/** Stops the active file block or active slot.
 * 
 * @param stopslot The slot number to stop. */
int BlockWriteServer::stopFileBlock(uint32_t stopslot)
{
   apr_atomic_set32(&(blockarray[stopslot].blockid), 0);
   apr_pool_destroy(blockarray[stopslot].localmp);
   apr_atomic_set32(&(blockarray[stopslot].active), 0);
   blockarray[stopslot].localmp = NULL;
}

/** When all of the packets have been assembled, this is called to
 * hash them and compare with the given hash, writing the result out
 * to a range of bytes in the file if successful. On failure, this
 * resets download of the hashed block.
 * 
 * @param slotID In what active slot the hash block is residing.
 * 
 * @param hashblockid The index of the hashed block inside this
 * slot. */
void BlockWriteServer::calculateCompletedHashedBlock(uint32_t slotID, uint32_t hashblockid)
{
   int hash_length = ((blockarray[slotID].blocklength - (hashblockid * HASHING_BLOCK_SIZE)) < HASHING_BLOCK_SIZE) ?
      (blockarray[slotID].blocklength - (hashblockid * HASHING_BLOCK_SIZE)) : HASHING_BLOCK_SIZE;
   // Needs to be padded with zeroes at the end beyond the actual
   // message length otherwise if the length is HASHING_BLOCK_SIZE
   // it's fine. (Someone could have overwritten this maliciously to
   // cause the download to never finish otherwise)
   for (int s = hash_length; s < HASHING_BLOCK_SIZE; s++)
      blockarray[slotID].hashblock[hashblockid].hashblockdata[s] = 0x00;
   // Put a SHA-1 hash of the data into the checksum part of the hash
   // block.
   Hash::getBinarySHA1Hash((const char *)(blockarray[slotID].hashblock[hashblockid].hashblockdata),
      HASHING_BLOCK_SIZE, blockarray[slotID].hashblock[hashblockid].checksum);
   // We don't need any locking because the full bit vector and full
   // complement of packets protects us.
   if (memcmp(blockarray[slotID].hashblock[hashblockid].checksum,
         &(blockarray[slotID].hashes[hashblockid*SHA1_DIGEST_LENGTH]),
         sizeof(uint8_t)*SHA1_DIGEST_LENGTH) == 0)
   {
      // Push the hashed block to the disk
      commitFileBlock(blockarray[slotID].filename, blockarray[slotID].offset + hashblockid*HASHING_BLOCK_SIZE,
         (const char *)(blockarray[slotID].hashblock[hashblockid].hashblockdata), hash_length, blockarray[slotID].localmp);
      // Set the bit saying the hash is fine
      BitVector::setBitAtomic(blockarray[slotID].verified_bit_vector, hashblockid);
   } else
   {
      // Reset the bit vector for this hashed block - block hash
      // failed
      for (int j = 0; j < ((HASHING_BLOCK_SIZE / PACKET_BLOCK_SIZE) / (sizeof(uint32_t) * 8)); j++)
         blockarray[slotID].hashblock[hashblockid].filled_bit_vector[j] = 0;
      // Reset any zero padding to previous fake complete state.
      // Block out the filled vector parts that are outside the block
      // with 1.
      for (int j = ((hash_length / PACKET_BLOCK_SIZE) + 1); j < (HASHING_BLOCK_SIZE / PACKET_BLOCK_SIZE); j++)
         BitVector::setBitFast(blockarray[slotID].hashblock[hashblockid].filled_bit_vector, j);
      if ((hash_length < HASHING_BLOCK_SIZE) && ((hash_length % PACKET_BLOCK_SIZE) == 0))
         BitVector::setBitFast(blockarray[slotID].hashblock[hashblockid].filled_bit_vector, (hash_length / PACKET_BLOCK_SIZE));
      // Also add the appropriate number of fake packets to the reset
      // packet count
      apr_atomic_set32(&(blockarray[slotID].hashblock[hashblockid].committedblocks),
         (HASHING_BLOCK_SIZE - hash_length) / PACKET_BLOCK_SIZE);
   }
}

/** Returns a new bit vector copy of the current state of the download
 * of this block.
 * 
 * @param blockID The block ID of the block whose bit vector state is
 * to be obtained.
 * 
 * @return A new array filled with a bit vector of active slot
 * state. */
uint32_t *BlockWriteServer::getActiveBlockFillVector(uint32_t blockID)
{
   if (!blockID)
      return NULL;
   // Find the slotID that corresponds to this blockID
   int slotID = -1;
   for (int i = 0; i < nomemfileblocks; i++)
   {
      if ((blockarray[i].active) && (blockID == blockarray[i].blockid))
      {
         slotID = i;
         break;
      }
   }
   if (slotID == -1)
      return NULL;
   // Copy all the bit vectors from all hash blocks into one long bit
   // vector (hopefully we can stream it to the client)
   uint32_t *returned_filled_bit_vector = (uint32_t *)malloc(
      (ACTIVE_BLOCK_SIZE / PACKET_BLOCK_SIZE) / (sizeof(uint32_t) * 8));
   for (int i = 0; i < (ACTIVE_BLOCK_SIZE / HASHING_BLOCK_SIZE); i++)
   {
      memcpy(&(returned_filled_bit_vector[i*((HASHING_BLOCK_SIZE / PACKET_BLOCK_SIZE) / (sizeof(uint32_t) * 8))]),
         blockarray[slotID].hashblock[i].filled_bit_vector, ((HASHING_BLOCK_SIZE / PACKET_BLOCK_SIZE) / (sizeof(uint32_t) * 8)));
   }
   // Actually, we probably want to return a filled_bit_vector here!
   return returned_filled_bit_vector;
}

/** Is called to write a block of data out to the given file.
 * 
 * @param filename The name of the file to write the block to.
 * 
 * @param offset The offset in the file to start the writing at.
 * 
 * @param data The data to write to the file at this point.
 * 
 * @param datalength The length of the data to write to the file here.
 * 
 * @param t_mp A memory pool from which to get the file opening
 * memory.
 * 
 * @return Zero on success, and a non-zero error code on failure. */
int BlockWriteServer::commitFileBlock(const char *filename, int64_t offset, const char *data, int datalength, apr_pool_t *t_mp)
{
   APRAutoLock autofileslock(fileslock);
   if (!filename)
   {
      errprint("file name NULL");
      return BLOCK_WRITE_SERVER__FILE_NAME_ERROR;
   }
   // Open the target file
   apr_status_t status;
   apr_file_t *filepntr;
   status = APRFile::openWrapper(&filepntr, filename, APR_FOPEN_BINARY|APR_FOPEN_WRITE|APR_FOPEN_BUFFERED, APR_FPROT_OS_DEFAULT, t_mp);
   // If we can't open it, it probably doesn't exist. Set
   // inactive. Return error.
   if (status != APR_SUCCESS)
   {
      errprint("file '%s' unallocated, not allowed", filename);
      return BLOCK_WRITE_SERVER__FILE_UNALLOCATED;
   }
   // Seek to the appropriate point in the file
   apr_off_t thisseekd = offset;
   status = apr_file_seek(filepntr, APR_SET, &thisseekd);
   CheckAPRError(status);
   if (status != APR_SUCCESS)
   {
      errprint("could not seek to %" APR_OFF_T_FMT " in file '%s'", (apr_off_t)(offset), filename);
      return BLOCK_WRITE_SERVER__FILE_SEEK_ERROR;
   }
   apr_size_t write_length = datalength;
   // Write block to file
   status = apr_file_write_full(filepntr, data, datalength, &write_length);
   CheckAPRError(status);
   if (status != APR_SUCCESS)
   {
      errprint("could not write block bytes %" APR_OFF_T_FMT "-%" APR_OFF_T_FMT, (apr_off_t)(offset), (apr_off_t)(offset + datalength));
      return BLOCK_WRITE_SERVER__FILE_WRITE_ERROR;
   }
   status = apr_file_close(filepntr);
   CheckAPRError(status);
}

/** Writes a packet that has been received into the appropriate place.
 * This selects the slot ID, hashed block, writes the data and
 * increments the number of written packets if necessary. If the
 * number of written packets goes over the number necessary to
 * complete the scheme, we recompute the hash and call to write data
 * out to the file if there is a matching hash.
 * 
 * @param blockID The block ID for the current open block.
 * 
 * @param block_offset The offset into the currently open slot.
 * 
 * @param packetdata A pointer to the data inside the UDP packet. */
void BlockWriteServer::writePacket(uint16_t blockID, uint16_t block_offset, const char *packetdata)
{
   if ((block_offset % PACKET_BLOCK_SIZE) > 0)
   {
      errprint("packet block incorrectly aligned with destination");
      return;
   }
   if (!blockID)
   {
      errprint("received packet with invalid block ID");
      return;
   }
   // Find the slotID that corresponds to this blockID
   int slotID = -1;
   for (int i = 0; i < nomemfileblocks; i++)
   {
      if ((blockarray[i].active) && (blockID == (uint16_t)(blockarray[i].blockid & 0xFFFF)))
      {
         slotID = i;
         break;
      }
   }
   if (slotID == -1)
   {
      errprint("received packet with non-matching block ID");
      return;
   }
   if ((blockarray[slotID].blocklength) <= (block_offset))
   {
      errprint("block offset %d too large for block of size %d", (int)block_offset, (int)blockarray[slotID].blocklength);
      return;
   }
   int64_t hashblock_idx = block_offset / HASHING_BLOCK_SIZE;
   // If we have verified and committed a block here already, do
   // nothing.
   if (BitVector::getBitFast(blockarray[slotID].verified_bit_vector, hashblock_idx))
      return;
   int64_t packetblock_byteoffset = block_offset % HASHING_BLOCK_SIZE;
   // If we've counted enough packets to make it complete, return.
   if ((blockarray[slotID].hashblock[hashblock_idx].committedblocks) ==
         (HASHING_BLOCK_SIZE / PACKET_BLOCK_SIZE))
      return;
   // If we have committed and not invalidated a block here, then do
   // nothing.
   if (BitVector::getBitFast(blockarray[slotID].hashblock[hashblock_idx].filled_bit_vector,
         packetblock_byteoffset / PACKET_BLOCK_SIZE))
      return;
   // Set a pointer to the live data.
   char *datapntr = 
    (char *)(&(blockarray[slotID].hashblock[hashblock_idx].hashblockdata[packetblock_byteoffset]));
   // OKAY, we don't care about the rare race situation where multiple
   // threads memcpys could cause corruption - the hashes will fix
   // this
   memcpy(datapntr, packetdata, PACKET_BLOCK_SIZE);
   // If we set the bit in the filled_bit_vector before anyone else,
   // we should increment the packet count
   if (!(BitVector::setBitAtomic(blockarray[slotID].hashblock[hashblock_idx].filled_bit_vector,
         packetblock_byteoffset / PACKET_BLOCK_SIZE)))
   {
      // We were beaten to this packet by someone else!
      return;
   }
   apr_atomic_inc32(&(blockarray[slotID].packetcounter));
   apr_atomic_inc32(&(blockarray[slotID].hashblock[hashblock_idx].committedblocks));
   // This block should be complete - now to test it against the
   // hashes to see if it needs resetting and regetting
   if (blockarray[slotID].hashblock[hashblock_idx].committedblocks == (HASHING_BLOCK_SIZE / PACKET_BLOCK_SIZE))
      calculateCompletedHashedBlock(slotID, hashblock_idx);
}
